use anyhow::Result;
use std::net::SocketAddr;
use std::time::{Duration, Instant};
use std::sync::Arc;
use tokio::net::{TcpStream, TcpListener as TokioTcpListener};
use tokio::io::{AsyncReadExt, AsyncWriteExt, AsyncSeekExt, BufReader, BufWriter};
use tokio::sync::Semaphore;
use shared::types::Session;

/// 传输统计信息
#[derive(Debug, Clone)]
pub struct TransferStats {
    pub bytes_transferred: u64,
    pub duration: Duration,
    pub speed_bytes_per_sec: f64,
    pub start_time: Instant,
}

impl std::fmt::Display for TransferStats {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{} bytes transferred in {:.2}s at {:.2} bytes/s", 
               self.bytes_transferred, 
               self.duration.as_secs_f64(), 
               self.speed_bytes_per_sec)
    }
}

/// 传输模式
#[derive(Debug, Clone, Copy)]
pub enum TransferMode {
    Passive,
    Active,
}

/// 传输类型
#[derive(Debug, Clone, Copy)]
pub enum TransferType {
    Binary,
    Ascii,
}

pub struct DataTransferManager {
    // 连接池限制
    connection_semaphore: Arc<Semaphore>,
    // 缓冲区大小配置
    buffer_size: usize,
    // 零拷贝传输启用
    zero_copy_enabled: bool,
    // 传输统计
    total_transfers: Arc<std::sync::atomic::AtomicU64>,
    total_bytes: Arc<std::sync::atomic::AtomicU64>,
}

impl DataTransferManager {
    pub fn new() -> Self {
        Self::with_config(1024, 65536, true) // 默认1024个连接，64KB缓冲区，启用零拷贝
    }
    
    pub fn with_config(max_connections: usize, buffer_size: usize, zero_copy_enabled: bool) -> Self {
        Self {
            connection_semaphore: Arc::new(Semaphore::new(max_connections)),
            buffer_size,
            zero_copy_enabled,
            total_transfers: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            total_bytes: Arc::new(std::sync::atomic::AtomicU64::new(0)),
        }
    }
    
    /// 获取连接许可
    pub async fn acquire_connection(&self) -> Result<tokio::sync::SemaphorePermit<'_>> {
        Ok(self.connection_semaphore.acquire().await?)
    }
    
    /// 获取传输统计
    pub fn get_stats(&self) -> (u64, u64) {
        (
            self.total_transfers.load(std::sync::atomic::Ordering::Relaxed),
            self.total_bytes.load(std::sync::atomic::Ordering::Relaxed),
        )
    }
    
    /// 创建被动模式数据连接
    pub async fn create_passive_connection(&self, _session: &Session) -> Result<(TcpStream, SocketAddr)> {
        // 创建监听器
        let listener = TokioTcpListener::bind("0.0.0.0:0").await?;
        let _local_addr = listener.local_addr()?;
        
        // 等待客户端连接
        let (stream, peer_addr) = listener.accept().await?;
        
        Ok((stream, peer_addr))
    }
    
    /// 创建主动模式数据连接
    pub async fn create_active_connection(&self, client_addr: SocketAddr, data_port: u16) -> Result<TcpStream> {
        let data_addr = SocketAddr::new(client_addr.ip(), data_port);
        let stream = TcpStream::connect(data_addr).await?;
        Ok(stream)
    }
    
    /// 上传文件（优化版本）
    pub async fn upload_file(
        &self,
        data_stream: &mut tokio::io::BufReader<tokio::net::TcpStream>,
        file_path: &std::path::Path,
        resume_offset: Option<u64>,
        transfer_type: TransferType,
    ) -> Result<TransferStats> {
        let start_time = Instant::now();
        let mut file = tokio::fs::OpenOptions::new()
            .create(true)
            .write(true)
            .open(file_path)
            .await?;
        
        // 如果指定了恢复偏移量，跳过已传输的部分
        if let Some(offset) = resume_offset {
            file.seek(std::io::SeekFrom::Start(offset)).await?;
        }
        
        // 使用更大的缓冲区
        let mut buffer = vec![0u8; self.buffer_size];
        let mut total_bytes = 0u64;
        
        // 创建带缓冲的文件写入器
        let mut file_writer = BufWriter::with_capacity(self.buffer_size, file);
        
        loop {
            let bytes_read = data_stream.read(&mut buffer).await?;
            if bytes_read == 0 {
                break; // 连接关闭
            }
            
            match transfer_type {
                TransferType::Binary => {
                    if self.zero_copy_enabled {
                        // 零拷贝写入
                        file_writer.write_all(&buffer[..bytes_read]).await?;
                    } else {
                        file_writer.write_all(&buffer[..bytes_read]).await?;
                    }
                }
                TransferType::Ascii => {
                    // ASCII模式需要处理换行符转换
                    let mut ascii_data = Vec::with_capacity(bytes_read * 2);
                    for &byte in &buffer[..bytes_read] {
                        if byte == b'\n' {
                            ascii_data.push(b'\r');
                            ascii_data.push(b'\n');
                        } else {
                            ascii_data.push(byte);
                        }
                    }
                    file_writer.write_all(&ascii_data).await?;
                }
            }
            
            total_bytes += bytes_read as u64;
        }
        
        file_writer.flush().await?;
        
        // 更新统计
        self.total_transfers.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.total_bytes.fetch_add(total_bytes, std::sync::atomic::Ordering::Relaxed);
        
        let duration = start_time.elapsed();
        let speed = self.calculate_speed(total_bytes, duration.as_millis() as u64);
        
        Ok(TransferStats {
            bytes_transferred: total_bytes,
            duration,
            speed_bytes_per_sec: speed,
            start_time,
        })
    }
    
    /// 下载文件（优化版本）
    pub async fn download_file(
        &self,
        data_stream: &mut tokio::io::BufWriter<tokio::net::TcpStream>,
        file_path: &std::path::Path,
        resume_offset: Option<u64>,
        transfer_type: TransferType,
    ) -> Result<TransferStats> {
        let start_time = Instant::now();
        let mut file = tokio::fs::File::open(file_path).await?;
        
        // 如果指定了恢复偏移量，跳过已传输的部分
        if let Some(offset) = resume_offset {
            file.seek(std::io::SeekFrom::Start(offset)).await?;
        }
        
        // 使用更大的缓冲区
        let mut buffer = vec![0u8; self.buffer_size];
        let mut total_bytes = 0u64;
        
        // 创建带缓冲的文件读取器
        let mut file_reader = BufReader::with_capacity(self.buffer_size, file);
        
        loop {
            let bytes_read = file_reader.read(&mut buffer).await?;
            if bytes_read == 0 {
                break; // 文件结束
            }
            
            match transfer_type {
                TransferType::Binary => {
                    if self.zero_copy_enabled {
                        // 零拷贝传输
                        data_stream.write_all(&buffer[..bytes_read]).await?;
                    } else {
                        data_stream.write_all(&buffer[..bytes_read]).await?;
                    }
                }
                TransferType::Ascii => {
                    // ASCII模式需要处理换行符转换
                    let mut ascii_data = Vec::with_capacity(bytes_read);
                    for &byte in &buffer[..bytes_read] {
                        if byte == b'\r' && ascii_data.last() == Some(&b'\n') {
                            // 跳过多余的\r
                            continue;
                        }
                        ascii_data.push(byte);
                    }
                    data_stream.write_all(&ascii_data).await?;
                }
            }
            
            total_bytes += bytes_read as u64;
        }
        
        data_stream.flush().await?;
        
        // 更新统计
        self.total_transfers.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.total_bytes.fetch_add(total_bytes, std::sync::atomic::Ordering::Relaxed);
        
        let duration = start_time.elapsed();
        let speed = self.calculate_speed(total_bytes, duration.as_millis() as u64);
        
        Ok(TransferStats {
            bytes_transferred: total_bytes,
            duration,
            speed_bytes_per_sec: speed,
            start_time,
        })
    }
    
    /// 传输目录列表
    pub async fn transfer_directory_list(
        &self,
        data_stream: &mut tokio::io::BufWriter<tokio::net::TcpStream>,
        directory_path: &std::path::Path,
    ) -> Result<TransferStats> {
        let start_time = Instant::now();
        let mut dir = tokio::fs::read_dir(directory_path).await?;
        let mut total_bytes = 0u64;
        
        while let Some(entry) = dir.next_entry().await? {
            let path = entry.path();
            let metadata = entry.metadata().await?;
            
            let entry_type = if metadata.is_dir() { "d" } else { "-" };
            let permissions = "rwxrwxrwx"; // 简化权限显示
            let size = metadata.len();
            let modified = metadata.modified()?;
            
            // 格式化时间
            let time_str = chrono::DateTime::<chrono::Utc>::from(modified)
                .format("%b %d %H:%M")
                .to_string();
            
            let filename = path.file_name()
                .and_then(|n| n.to_str())
                .unwrap_or("");
            
            // 生成类似 ls -l 的输出
            let line = format!(
                "{} {} {} {} {} {} {} {}\r\n",
                entry_type,
                permissions,
                "1", // 硬链接数
                "user", // 用户
                "group", // 组
                size,
                time_str,
                filename
            );
            
            data_stream.write_all(line.as_bytes()).await?;
            total_bytes += line.len() as u64;
        }
        
        data_stream.flush().await?;
        
        let duration = start_time.elapsed();
        let speed = self.calculate_speed(total_bytes, duration.as_millis() as u64);
        
        Ok(TransferStats {
            bytes_transferred: total_bytes,
            duration,
            speed_bytes_per_sec: speed,
            start_time,
        })
    }
    
    /// 计算传输速度
    pub fn calculate_speed(&self, bytes_transferred: u64, duration_ms: u64) -> f64 {
        if duration_ms == 0 {
            return 0.0;
        }
        
        let duration_secs = duration_ms as f64 / 1000.0;
        bytes_transferred as f64 / duration_secs
    }

    /// 格式化传输速度
    pub fn format_speed(&self, speed_bytes_per_sec: f64) -> String {
        if speed_bytes_per_sec < 1024.0 {
            format!("{:.2} B/s", speed_bytes_per_sec)
        } else if speed_bytes_per_sec < 1024.0 * 1024.0 {
            format!("{:.2} KB/s", speed_bytes_per_sec / 1024.0)
        } else {
            format!("{:.2} MB/s", speed_bytes_per_sec / (1024.0 * 1024.0))
        }
    }

    /// 格式化文件大小
    pub fn format_size(&self, size_bytes: u64) -> String {
        if size_bytes < 1024 {
            format!("{} B", size_bytes)
        } else if size_bytes < 1024 * 1024 {
            format!("{:.2} KB", size_bytes as f64 / 1024.0)
        } else if size_bytes < 1024 * 1024 * 1024 {
            format!("{:.2} MB", size_bytes as f64 / (1024.0 * 1024.0))
        } else {
            format!("{:.2} GB", size_bytes as f64 / (1024.0 * 1024.0 * 1024.0))
        }
    }

    /// 检查文件是否支持断点续传
    pub async fn supports_resume(&self, file_path: &std::path::Path) -> bool {
        tokio::fs::metadata(file_path).await.is_ok()
    }

    /// 获取文件大小
    pub async fn get_file_size(&self, file_path: &std::path::Path) -> Result<u64> {
        let metadata = tokio::fs::metadata(file_path).await?;
        Ok(metadata.len())
    }

    /// 验证传输完整性（简单的字节数检查）
    pub fn verify_transfer(&self, expected_size: u64, actual_size: u64) -> bool {
        expected_size == actual_size
    }
    
    /// 批量传输文件（优化版本）
    pub async fn batch_transfer_files(
        &self,
        transfers: Vec<(String, String, TransferType)>, // (source, dest, type)
    ) -> Result<Vec<TransferStats>> {
        let mut results = Vec::new();
        
        // 使用并发传输
        let semaphore = Arc::new(Semaphore::new(10)); // 最多10个并发传输
        
        let mut handles = Vec::new();
        
        for (source, dest, _transfer_type) in transfers {
            let semaphore = semaphore.clone();
            let manager = self.clone();
            
            let handle = tokio::spawn(async move {
                let _permit = semaphore.acquire().await.unwrap();
                
                // 根据传输类型选择上传或下载
                if std::path::Path::new(&source).exists() {
                    // 文件存在，执行下载
                    let file = tokio::fs::File::open(&source).await?;
                    let mut reader = BufReader::with_capacity(manager.buffer_size, file);
                    let mut writer = BufWriter::with_capacity(manager.buffer_size, tokio::fs::File::create(&dest).await?);
                    
                    let mut buffer = vec![0u8; manager.buffer_size];
                    let mut total_bytes = 0u64;
                    let start_time = Instant::now();
                    
                    loop {
                        let bytes_read = reader.read(&mut buffer).await?;
                        if bytes_read == 0 {
                            break;
                        }
                        
                        writer.write_all(&buffer[..bytes_read]).await?;
                        total_bytes += bytes_read as u64;
                    }
                    
                    writer.flush().await?;
                    
                    let duration = start_time.elapsed();
                    let speed = manager.calculate_speed(total_bytes, duration.as_millis() as u64);
                    
                    Ok(TransferStats {
                        bytes_transferred: total_bytes,
                        duration,
                        speed_bytes_per_sec: speed,
                        start_time,
                    })
                } else {
                    Err(anyhow::anyhow!("Source file not found: {}", source))
                }
            });
            
            handles.push(handle);
        }
        
        // 等待所有传输完成
        for handle in handles {
            match handle.await? {
                Ok(stats) => results.push(stats),
                Err(e) => return Err(e),
            }
        }
        
        Ok(results)
    }
    
    /// 流式传输（用于大文件）
    pub async fn stream_transfer(
        &self,
        source: &std::path::Path,
        dest: &std::path::Path,
        chunk_size: usize,
    ) -> Result<TransferStats> {
        let start_time = Instant::now();
        let mut source_file = tokio::fs::File::open(source).await?;
        let mut dest_file = tokio::fs::File::create(dest).await?;
        
        let mut buffer = vec![0u8; chunk_size];
        let mut total_bytes = 0u64;
        
        loop {
            let bytes_read = source_file.read(&mut buffer).await?;
            if bytes_read == 0 {
                break;
            }
            
            dest_file.write_all(&buffer[..bytes_read]).await?;
            total_bytes += bytes_read as u64;
            
            // 定期刷新以确保数据写入磁盘
            if total_bytes % (chunk_size * 10) as u64 == 0 {
                dest_file.flush().await?;
            }
        }
        
        dest_file.flush().await?;
        
        let duration = start_time.elapsed();
        let speed = self.calculate_speed(total_bytes, duration.as_millis() as u64);
        
        Ok(TransferStats {
            bytes_transferred: total_bytes,
            duration,
            speed_bytes_per_sec: speed,
            start_time,
        })
    }
}

impl Clone for DataTransferManager {
    fn clone(&self) -> Self {
        Self {
            connection_semaphore: self.connection_semaphore.clone(),
            buffer_size: self.buffer_size,
            zero_copy_enabled: self.zero_copy_enabled,
            total_transfers: self.total_transfers.clone(),
            total_bytes: self.total_bytes.clone(),
        }
    }
}
